package cn.com.greatwall.kafka.util;

import java.io.Closeable;
import java.io.IOException;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.Header;

/**
 * @Author herw
 * @Time 2021-04-21 13:05:40
 * @Version 1.0
 * @Description: TODO(用一句话描述该文件做什么)
 */
public class Producers<K, V> implements Closeable {

    private KafkaProducer<K, V> producer;

    /**
     * props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProducerProperties.getBootstrap());
     * props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
     * props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
     * props.put("acks", "all");
     * props.put("retries", "0");
     * props.put("batch.size", "16384");
     * 
     */
    public Producers(Properties properties) {
        this.producer = new KafkaProducer<K, V>(properties);
    }

    /**
     * 同步发送
     * 
     * @param topicName
     * @param k
     * @param v
     * @throws InterruptedException
     * @throws ExecutionException
     */
    public void send(String topicName, K k, V v) throws InterruptedException, ExecutionException {
        this.producer.send(new ProducerRecord<K, V>(topicName, k, v)).get();
    }

    /**
     * 异步发送
     * 
     * @param topicName
     * @param partition
     * @param timestamp
     * @param key
     * @param value
     * @param headers
     * @throws InterruptedException
     * @throws ExecutionException
     */
    public void send(String topicName, Integer partition, Long timestamp, K key, V value, Iterable<Header> headers)
            throws InterruptedException, ExecutionException {
        this.producer.send(new ProducerRecord<K, V>(topicName, partition, timestamp, key, value, headers)).get();
    }

    public void flush() {
        this.producer.flush();
    }

    @Override
    public void close() throws IOException {
        this.producer.close();
    }
}
